package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/9/13
 * desc:
 **/
public class OrdersDemo {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    order_id       BIGINT,\n" +
                "    order_date      TIMESTAMP(3),\n" +
                "    customer_name   STRING,\n" +
                "    price   FLOAT,\n" +
                "    product_id INT,\n" +
                "    order_status TINYINT\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
                "  'number-of-rows' = '12', \n" +
//                "  'rows-per-second' = '1', \n" +
                "  'fields.order_id.min' = '10', \n" +
                "  'fields.order_id.max' = '200', \n" +
                "  'fields.customer_name.length' = '5', \n" +
                "  'fields.product_id.min' = '10', \n" +
                "  'fields.product_id.max' = '100', \n" +
                "  'fields.order_status.min' = '1', \n" +
                "  'fields.order_status.max' = '10', \n" +
                "  'fields.price.min' = '10', \n" +
                "  'fields.price.max' = '100'\n" +
                ")");

                //Sink
        tableEnv.executeSql("CREATE TABLE sinkTable (" +
                "    order_id       BIGINT,\n" +
                "    order_date      TIMESTAMP(3),\n" +
                "    customer_name   STRING,\n" +
                "    price   FLOAT,\n" +
                "    product_id INT,\n" +
                "    order_status TINYINT\n" +
                "   ) " +
                "   WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'orders',\n" +
                "  'properties.bootstrap.servers' = '192.168.102.154:29092',\n" +
                "  'properties.allow.auto.create.topics' = 'true',\n" +
                "  'sink.parallelism' = '5',\n" +
                "  'format' = 'json'\n" +
//                "  'format' = 'debezium-json'\n" +
                ")");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                        "LIKE sourceTable (EXCLUDING ALL)");

        //Insert
        tableEnv.executeSql("insert into sinkTable select * from sourceTable");


    }
}
